package com.ds.lens.flink.monitor.eventsinkjob.function;

import com.ds.lens.data.common.client.LensMonitorClient;
import com.ds.lens.data.common.client.LensResponseData;
import com.ds.lens.data.common.client.vo.LensMetricInVO;
import com.ds.lens.data.common.constant.*;
import com.ds.lens.data.common.document.DataPoint;
import com.ds.lens.data.common.document.Event;
import com.ds.lens.data.common.document.dto.MonitorEvent;
import com.ds.lens.flink.monitor.MonitorJobContext;
import com.ds.lens.flink.monitor.common.DataPointComparator;
import com.ds.lens.flink.monitor.common.LogUtils;
import com.ds.lens.flink.monitor.eventsinkjob.function.notify.MonitorEventNotifier;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.springframework.util.CollectionUtils;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import static com.ds.lens.data.common.constant.Constants.METRIC_VALUE;
import static com.ds.lens.data.common.constant.Constants.TIME_STAMP;

@Slf4j
public class MonitorEventNotifyFunction extends ProcessFunction<MonitorEvent, Event> {
    private static final String DATE_PATTERN = "yyyy-MM-dd HH:mm:ss";
    private String lensEventWebUrl = "/monitor/%s?eventId=%s&from=eventList";

    @Override
    public void processElement(MonitorEvent value, Context ctx, Collector<Event> out) throws Exception {
        String ticketId = null;
        long pastDataStartTime = 0;
        if (value.getIsNotify()) {
            if (EventSourceEnum.LENS_EVENT.equals(value.getEventSourceEnum())) {
                //替换变量
                replaceWildcard(value);
                ticketId = sendNotice(value);
                pastDataStartTime = updateMetrics(value);
            } else {
                // 只发通知
                ticketId = sendNotice(value);
            }
        }

        LogUtils.debug("generate event by monitorevent={}", value);
        Event result = Event.builder()
                .uid(value.getUid())
                .detectionMethod(value.getDetectionMethod())
                .monitorId(value.getMonitorKey() != null ? value.getMonitorKey().getMonitorId() : null)
                .version((value.getMonitorKey() != null && value.getMonitorKey().getVersion() != null) ? value.getMonitorKey().getVersion().longValue() : null)
                .name(value.getMonitorName())
                .monitorCreateTime(value.getMonitorCreateTime())
                .pastDataStartTime(pastDataStartTime)
                .threshold(value.getLevelThresholdMap())
                .state(value.getState().name())
                .renotify(null != value.getTriggered() ? value.getTriggered().getRenotify() : false)
                .tags(value.getTags())
                .expression(value.getExpression())
                .notify(value.getNotify())
                .triggered(value.getTriggered())
                .timeStamp(value.getTimeStamp())
                .ticket(ticketId == null ? "" : ticketId)
                .interval(value.getInterval())
                .queryDelayTime(value.getQueryDelayTime())
                .retriveDuration(value.getRetriveDuration())
                .metricName(value.getMetricName())
                .metricAggr(value.getMetricAggr())
                .scope(value.getScope())
                .integrationId(value.getIntegrationId())
                .metricFilter(value.getMetricFilter())
                .metricBy(value.getMetricBy())
                .appId(value.getAppId())
                .build();
        out.collect(result);
    }

    private String sendNotice(MonitorEvent value) {
        String ticketId = null;
        MonitorEventNotifier<MonitorEvent, String> monitorEventNotifier = MonitorJobContext.getInstance().getChannelMap()
                .get(value.getNotify().getChannel());
        if (monitorEventNotifier != null) {
            try {
                ticketId = doSend(value, monitorEventNotifier);
            } catch (Exception e) {
                log.error("Send error.", e);
            }
        } else {
            log.error("当前发送通道不可用，通道id：{}", value.getNotify().getChannel());
        }
        return ticketId;
    }

    /**
     * 发送错误重发一次
     *
     * @param value
     * @param monitorEventNotifier
     * @return
     * @throws Exception
     */
    private String doSend(MonitorEvent value, MonitorEventNotifier<MonitorEvent, String> monitorEventNotifier) throws Exception {
        try {
            return monitorEventNotifier.send(value);
        } catch (Exception e) {
            return monitorEventNotifier.send(value);
        }
    }

    private long updateMetrics(MonitorEvent value) {
        LensMonitorClient monitorClient = MonitorJobContext.getInstance().getMonitorClient();
        long startTime = value.getQueryEndTime() - 30 * value.getInterval() * 60000;
        long endTime = value.getQueryStartTime();
        if (startTime >= endTime) {
            return value.getQueryStartTime();
        }
        LensMetricInVO.LensMetricInVOBuilder inVOBuilder = LensMetricInVO.builder()
                .scope(value.getScope())
                .integrationId(value.getIntegrationId())
                .appId(value.getAppId())
                .metricName(value.getMetricName())
                .metricAggr(value.getMetricAggr())
                .endTime(endTime)
                .startTime(startTime)
                .metricBy(value.getMetricBy())
                .interval(value.getInterval());
//        Map<String, Object> filters = new HashMap<>();
//        MapUtils.merge(filters, value.getMetricFilter(), value.getGroupBys());
        inVOBuilder.metricFilter(value.getMetricFilter());
        LensMetricInVO inVO = inVOBuilder.build();
        LensResponseData<List<Map<String, Object>>> resp = monitorClient.queryMetrics(inVO);
        List<Map<String, Object>> results = resp.getData();
        LogUtils.debug("update metrics by MonitorEvent={}, LensMetricInVO={} and result is empty? {}",
                value, inVO, CollectionUtils.isEmpty(results));
        List<DataPoint> dataPoints = new ArrayList<>();
        if (!CollectionUtils.isEmpty(results)) {
            for (Map<String, Object> map : results) {
                Double metricValue = Double.valueOf(map.get(METRIC_VALUE).toString());
                String timeStr = map.get(TIME_STAMP).toString();
                Long timeStamp = Long.valueOf(new BigDecimal(timeStr).toPlainString());
                dataPoints.add(new DataPoint(metricValue, timeStamp, MonitorStateEnum.OK, null, null, null));
            }
        }
        List<DataPoint> triggered = value.getTriggered().getDataPoint();
        if (!CollectionUtils.isEmpty(triggered)) {
            for (DataPoint dataPoint : dataPoints) {
                if (!triggered.contains(dataPoint)) {
                    triggered.add(dataPoint);
                }
            }
        } else {
            value.getTriggered().setDataPoint(dataPoints);
        }

        value.getTriggered().getDataPoint().sort(new DataPointComparator());
        return startTime;
    }

    private void replaceWildcard(MonitorEvent event) {
        if (EventStateEnum.INFO.equals(event.getState()) || null == event.getTriggered()) {
            return;
        }
        //判断通知模板类型
        String content = getContentTemplateStr(event);
        //静态变量替换
        content = replaceStaticWildcard(event, content);
        //动态变量替换
        content = replaceDynamicWildcard(content, event.getGroupBys());
        content = replaceDynamicWildcard(content, event.getMetricFilter());

        event.getNotify().setContent(content);
    }

    private String getContentTemplateStr(MonitorEvent event) {
        String template = event.getNotify().getContent();
        String content = template;
        switch (event.getTriggered().getState()) {
            case NO_DATA:
                if (template.contains(WildcardEnum.TEMPLATE_NO_DATA_START.getWildcard()) && template.contains(WildcardEnum.TEMPLATE_NO_DATA_END.getWildcard())) {
                    content = template.substring(template.indexOf(WildcardEnum.TEMPLATE_NO_DATA_START.getWildcard()) + WildcardEnum.TEMPLATE_NO_DATA_START.getWildcard().length(),
                            template.indexOf(WildcardEnum.TEMPLATE_NO_DATA_END.getWildcard()));
                }
                break;
            case OK:
                if (template.contains(WildcardEnum.TEMPLATE_IS_OK_START.getWildcard()) && template.contains(WildcardEnum.TEMPLATE_IS_OK_END.getWildcard())) {
                    content = template.substring(template.indexOf(WildcardEnum.TEMPLATE_IS_OK_START.getWildcard()) + WildcardEnum.TEMPLATE_IS_OK_START.getWildcard().length(),
                            template.indexOf(WildcardEnum.TEMPLATE_IS_OK_END.getWildcard()));
                }
                break;
            case ALARM_LEVEL_1:
            case ALARM_LEVEL_2:
            case ALARM_LEVEL_3:
            case ALARM_LEVEL_4:
            case ALARM_LEVEL_5:
                if (template.contains(WildcardEnum.TEMPLATE_IS_ALERT_START.getWildcard()) && template.contains(WildcardEnum.TEMPLATE_IS_ALERT_END.getWildcard())) {
                    content = template.substring(template.indexOf(WildcardEnum.TEMPLATE_IS_ALERT_START.getWildcard()) + WildcardEnum.TEMPLATE_IS_ALERT_START.getWildcard().length(),
                            template.indexOf(WildcardEnum.TEMPLATE_IS_ALERT_END.getWildcard()));
                }
                break;
            default:
                //do nothing
        }
        return content;
    }

    private String replaceStaticWildcard(MonitorEvent event, String content) {
        if (null != event.getAppName()) {
            content = content.replaceAll(WildcardEnum.TEMPLATE_APP.getWildcard(), String.valueOf(event.getAppName()));
        } else {
            content = content.replaceAll(WildcardEnum.TEMPLATE_APP.getWildcard(), String.valueOf(event.getAppId()));
        }
        if (StringUtils.isNotEmpty(event.getMetricName())) {
            content = content.replaceAll(WildcardEnum.TEMPLATE_METRIC_NAME.getWildcard(), event.getMetricName().trim());
        }
        if (null != event.getAggrEnum()) {
            content = content.replaceAll(WildcardEnum.TEMPLATE_AGGREGATE.getWildcard(), event.getAggrEnum().name().trim());
        }
        if (null != event.getTriggered() && null != event.getTriggered().getState()) {
            content = content.replaceAll(WildcardEnum.TEMPLATE_ALARM_LEVEL.getWildcard(), event.getTriggered().getState().getDesc().trim());
        }
        if (null != event.getQueryEndTime()) {
            content = content.replaceAll(WildcardEnum.TEMPLATE_EVENT_END_TIME.getWildcard(), DateFormatUtils.format(event.getQueryEndTime(), DATE_PATTERN));
        }
        if (null != event.getQueryStartTime()) {
            content = content.replaceAll(WildcardEnum.TEMPLATE_EVENT_START_TIME.getWildcard(), DateFormatUtils.format(event.getQueryStartTime(), DATE_PATTERN));
        }
        if (org.apache.commons.collections.MapUtils.isNotEmpty(event.getScope())) {
            content = content.replaceAll(ApmWildcardEnum.TEMPLATE_APM_BUSINESS_DOMAIN.getWildcard(), String.valueOf(event.getScope().get(ApmWildcardEnum.TEMPLATE_APM_BUSINESS_DOMAIN.getApmMapping())));
            content = content.replaceAll(ApmWildcardEnum.TEMPLATE_APM_SERVICE.getWildcard(), String.valueOf(event.getScope().get(ApmWildcardEnum.TEMPLATE_APM_SERVICE.getApmMapping())));
        }
        if (StringUtils.isNotEmpty(event.getUid()) && null != event.getMonitorKey().getMonitorId()) {
            String urlStr = "<a href=\"" + MonitorJobContext.getInstance().getLensWebHost() + lensEventWebUrl + "\">告警详情</a>";
            urlStr = String.format(urlStr, event.getMonitorKey().getMonitorId(), event.getUid());
            content = content.replaceAll(WildcardEnum.TEMPLATE_DETAIL_INFO.getWildcard(), urlStr);
        }
        return content;
    }

    private String replaceDynamicWildcard(String content, Map<String, Object> objectMap) {
        if (org.apache.commons.collections.MapUtils.isNotEmpty(objectMap)) {
            for (Map.Entry<String, Object> entry : objectMap.entrySet()) {
                content = content.replaceAll("\\{\\{" + entry.getKey() + "\\}\\}", String.valueOf(entry.getValue()));
            }
        }
        return content;
    }

}
